package com.bw.wyq.function;

import com.alibaba.fastjson.JSONObject;
import com.bw.wyq.bean.TableProcessDim;
import com.bw.wyq.constant.Constant;
import com.bw.wyq.util.HbaseUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection connection;
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HbaseUtil.getHbaseConnect();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHBaseConn(connection);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        // 主流维度数据，并且已经去除没用的字段
        JSONObject f0 = value.f0;
        // 对应的配置表数据
        TableProcessDim f1 = value.f1;

        // 要往Hbase写入的数据
        JSONObject data = f0.getJSONObject("data");
        // 取出操作类型
        String type = f0.getString("type");

        // 往Hbase哪张表
        String sinkTable = f1.getSinkTable();
        // id dic_code
        String sinkRowKey = f1.getSinkRowKey();
        // 取出对应的维度数据主键的值
        String sinkRowKeyValue = data.getString(sinkRowKey);
        // 取出列族
        String sinkFamily = f1.getSinkFamily();
        if("delete".equals(type)){
            // 从Hbase删除这条数据
            HbaseUtil.deleteCells(connection, Constant.HBASE_NAMESPACE,sinkTable,sinkRowKeyValue);
        }else{
            HbaseUtil.putCells(connection,Constant.HBASE_NAMESPACE,sinkTable,sinkRowKeyValue,sinkFamily,data);
        }


    }
}
